package com.jugheadzhou.kafka.core.manager;

import com.jugheadzhou.kafka.core.enums.KafkaMessageTypeEnum;
import com.jugheadzhou.kafka.core.logger.AsyncKafkaLogger;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;

import javax.annotation.Resource;

/**
 * Kafka编程式发送消息
 *
 * @author jugheadzhou
 * @date 2021-09-23
 */
@Slf4j
public class KafkaMessageManager {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 异步发送kafka消息
     *
     * @param topic       主题
     * @param messageType 消息类型
     * @param data        业务数据
     */
    public void asyncSend(String topic, KafkaMessageTypeEnum messageType, Object data) {
        assert data != null;
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, AsyncKafkaLogger.wrapperKafkaMessage(messageType, data));
        future.addCallback(result -> {
            assert result != null;
            log.info("发送kafka消息成功，result={}", result);
        }, e -> log.error("发送kafka消息失败，原因是：", e));

        log.info("异步发送kafka消息结束");
    }

    /**
     * 同步发送kafka消息
     *
     * @param topic    主题
     * @param typeEnum 消息类型
     * @param data     业务数据
     */
    public void syncSend(String topic, KafkaMessageTypeEnum typeEnum, Object data) throws Exception {
        assert data != null;
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, AsyncKafkaLogger.wrapperKafkaMessage(typeEnum, data));
        SendResult<String, String> result = future.get();
        assert result != null;
        log.info("发送kafka消息成功，result={}", result);

        log.info("同步发送kafka消息结束");
    }


}
